package chan_stats

import (
	"encoding/json"
	"fmt"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"longmen/server/config/global"
	"longmen/server/internal/services/consumers/publisher"
	"longmen/server/pkg/common/api_models"
	"longmen/server/pkg/common/helper/const"
	"longmen/server/pkg/common/helper/timehelper"
	"longmen/server/pkg/common/lib/rabbitmq"
	"longmen/server/pkg/common/report"
	"longmen/server/pkg/common/services"
	"longmen/server/pkg/db/mysql/models/models_main"
	"longmen/server/pkg/db/redis/cache"
	"sync"
	"time"
)

var instance *ChannelStatistic
var once sync.Once
var funcMap map[publisher.ChannelStatType]rabbitmq.Process

func NewChannelStatistic(log *zap.SugaredLogger) *ChannelStatistic {
	once.Do(func() {
		instance = &ChannelStatistic{
			logger: log,
		}
		funcMap = map[publisher.ChannelStatType]rabbitmq.Process{
			publisher.LoginChannelStats:    instance.StatUserLoginChannelHandler,
			publisher.RegisterChannelStats: instance.StatUserRegisterChannelHandler,
			publisher.ActiveChannelStats:   instance.StatUserActiveChannelHandler,
			//publisher.LiveWatchStats:       instance.StatUserLiveWatchHandler,
			publisher.RegisterDyTransform: instance.DyTransformRegisterHandler,
		}
	})
	return instance
}

type ChannelStatistic struct {
	logger *zap.SugaredLogger
}

func (s *ChannelStatistic) Process(data []byte) bool {
	var resp publisher.ChannelStatMessage
	err := json.Unmarshal(data, &resp)
	if err != nil {
		return false
	}

	process, ok := funcMap[resp.ChanType]
	if !ok {
		return false
	}

	return process(resp.Msg)
}

// StatUserRegisterChannelHandler 注册渠道统计
func (s *ChannelStatistic) StatUserRegisterChannelHandler(msg []byte) bool {
	s.logger.Debug("渠道号挂载 收到订阅")
	var today = time.Now()
	var todayString = time.Now().Format("2006-01-02")

	var user models_main.User
	err := json.Unmarshal(msg, &user)
	if err != nil {
		s.logger.Errorf("渠道号挂载 SubMessage StatUserRegisterChannelHandler err [%+v]", err)
		return false
	}

	var exist int64

	tx := global.GetX().Begin()
	err = tx.Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, todayString).Count(&exist).Update("register", gorm.Expr("register+1")).Error

	if err != nil {
		tx.Rollback()
		s.logger.Debug("渠道号挂载 StatUserRegisterChannelHandler Error: ", err)
		return false
	}
	s.logger.Errorf("渠道号挂载 exist：%v", exist)
	if exist == 0 {
		var channel models_main.RegSource
		global.GetR().Model(&models_main.RegSource{}).Where("source_key=?", user.RegSourceId).Order("id asc").
			Limit(1).
			Find(&channel)

		r := &models_main.StatUser{
			SourceKey:  user.RegSourceId,
			SourceName: channel.SourceName,
			Register:   1,
			Date:       today,
		}

		err = tx.Model(&models_main.StatUser{}).Create(&r).Error
		if err != nil {
			tx.Rollback()
			return false
		}
	}

	tx.Commit()
	s.logger.Errorf("渠道号挂载 完成：%v", user)
	s.StatUserLoginChannelHandler(msg)
	return true
}

// StatUserLoginChannelHandler 登录渠道数统计
func (s *ChannelStatistic) StatUserLoginChannelHandler(msg []byte) bool {
	var today = time.Now()
	var todayString = time.Now().Format("2006-01-02")

	var user models_main.User
	json.Unmarshal(msg, &user)
	// 查询昨日登录记次
	var lastDayLoggin = 0
	err := global.GetX().Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, today.Add(-time.Hour*24).Format("2006-01-02")).Pluck("loggin_user", &lastDayLoggin).Error
	if err != nil {
		s.logger.Debug("StatUserLoginChannelHandler Error: ", err)
	}

	err = global.GetRds().Set(fmt.Sprintf(_const.USER_IP_SOURCE, user.RegistIp), time.Now().Unix(), time.Second*time.Duration(timehelper.DateSkewing(1)-time.Now().Unix())).Err()
	if err != nil {
		s.logger.Debug("Stat User Ip Source Error: ", err)
	}

	rcache, _ := global.GetRds().Get(fmt.Sprintf(_const.REDIS_STAT_USER_LOGIN, todayString, user.Id, user.RegSourceId)).Result()
	if rcache == "" {
		var channel models_main.RegSource
		global.GetR().Model(&models_main.RegSource{}).Where("source_key=?", user.RegSourceId).Order("id desc").
			Limit(1).
			Find(&channel)

		tx := global.GetX().Begin()
		var exist int64
		err := tx.Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, todayString).Count(&exist).Updates(map[string]interface{}{"loggin_user": gorm.Expr("loggin_user+1"), "last_day_loggin": lastDayLoggin}).Error
		if err != nil {
			tx.Rollback()
			s.logger.Debug("StatUserLoginChannelHandler Error: ", err)
			return false
		}

		if exist == 0 {
			r := &models_main.StatUser{
				SourceKey:     user.RegSourceId,
				SourceName:    channel.SourceName,
				LogginUser:    1,
				Date:          today,
				LastDayLoggin: lastDayLoggin,
			}

			err = tx.Model(&models_main.StatUser{}).Create(&r).Error
			if err != nil {
				s.logger.Debug("StatUserLoginChannelHandler Error: ", err)
				tx.Rollback()
				return false
			}
		}

		tx.Commit()

		global.GetRds().Set(fmt.Sprintf(_const.REDIS_STAT_USER_LOGIN, todayString, user.Id, user.RegSourceId), 1, time.Second*86400*2)
	}

	s.statUserNextdayChannelHandler(user)
	s.statUserValidChannelHandler(user)
	return true
}

// statUserNextdayChannelHandler 留存
func (s *ChannelStatistic) statUserNextdayChannelHandler(user models_main.User) {
	var today = time.Now()
	var todayString = time.Now().Format("2006-01-02")

	rcache, _ := global.GetRds().Get(fmt.Sprintf(_const.REDIS_STAT_USER_LOGIN, today.Add(-time.Hour*24).Format("2006-01-02"), user.Id, user.RegSourceId)).Result()
	alreadyDone, _ := global.GetRds().Get(fmt.Sprintf(_const.REDIS_STAT_USER_NEXTDAY, todayString, user.Id, user.RegSourceId)).Result()
	if rcache != "" && alreadyDone == "" {
		var channel models_main.RegSource
		global.GetR().Model(&models_main.RegSource{}).Where("source_key=?", user.RegSourceId).Order("id asc").
			Limit(1).
			Find(&channel)

		tx := global.GetX().Begin()

		var exist int64
		err := tx.Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, todayString).Count(&exist).Update("next_day_user", gorm.Expr("next_day_user+1")).Error
		if err != nil {
			tx.Rollback()
			s.logger.Debug("statUserNextdayChannelHandler Error: ", err)
			return
		}

		if exist == 0 {
			r := &models_main.StatUser{
				SourceKey:   user.RegSourceId,
				SourceName:  channel.SourceName,
				NextDayUser: 1,
				Date:        today,
			}

			err = tx.Model(&models_main.StatUser{}).Create(&r).Error
			if err != nil {
				tx.Rollback()
				return
			}
		}

		tx.Commit()

		global.GetRds().Set(fmt.Sprintf(_const.REDIS_STAT_USER_LOGIN, todayString, user.Id, user.RegSourceId), 1, time.Second*86400*2)
		global.GetRds().Set(fmt.Sprintf(_const.REDIS_STAT_USER_NEXTDAY, todayString, user.Id, user.RegSourceId), 1, time.Second*86400*2)
	}

}

// statUserValidChannelHandler 有效用户
func (s *ChannelStatistic) statUserValidChannelHandler(user models_main.User) {
	var today = time.Now()
	var todayString = time.Now().Format("2006-01-02")

	var userLog models_main.UserLog
	err := global.GetX().Model(&models_main.UserLog{}).Where("uid=?", user.Id).Order("id asc").
		Limit(1).
		Find(&userLog).Error
	if err != nil {
		s.logger.Debug("statUserValidChannelHandler Error: ", err)
	}

	if err == gorm.ErrRecordNotFound {
		return
	}

	rcache, _ := global.GetRds().Get(fmt.Sprintf(_const.REDIS_STAT_USER_VALID, todayString, user.Id, user.RegSourceId)).Result()

	if rcache == "" && userLog.CreateTime.Format("2006-01-02") != todayString {
		var channel models_main.RegSource
		global.GetR().Model(&models_main.RegSource{}).Where("source_key=?", user.RegSourceId).Order("id asc").
			Limit(1).
			Find(&channel)

		tx := global.GetX().Begin()
		var exist int64
		err := tx.Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, todayString).Count(&exist).Update("valid_user", gorm.Expr("valid_user+1")).Error
		if err != nil {
			tx.Rollback()
			s.logger.Debug("StatUserRegisterChannelHandler Error: ", err)
			return
		}

		if exist == 0 {
			r := &models_main.StatUser{
				SourceKey:  user.RegSourceId,
				SourceName: channel.SourceName,
				ValidUser:  1,
				Date:       today,
			}

			err = tx.Model(&models_main.StatUser{}).Create(&r).Error
			if err != nil {
				tx.Rollback()
				return
			}

		}

		tx.Commit()
		global.GetRds().Set(fmt.Sprintf(_const.REDIS_STAT_USER_VALID, todayString, user.Id, user.RegSourceId), 1, time.Second*86400*2)
	}
}

// StatUserActiveChannelHandler 活跃用户
func (s *ChannelStatistic) StatUserActiveChannelHandler(msg []byte) bool {
	return true
	var today = time.Now()
	var todayString = time.Now().Format("2006-01-02")

	var user models_main.User
	err := json.Unmarshal(msg, &user)
	if err != nil {
		global.Log.Errorf("StatUserActiveChannelHandler err:%v", err)
		return false
	}
	key := fmt.Sprintf(_const.REDIS_STAT_USER_ACTIVE, todayString, user.Id, user.RegSourceId)
	rcache, _ := global.GetRds().Get(key).Result()
	if rcache == "" {
		var channel models_main.RegSource
		global.GetR().Model(&models_main.RegSource{}).Where("source_key=?", user.RegSourceId).Order("id asc").
			Limit(1).
			Find(&channel)

		tx := global.GetX().Begin()
		var exist int64
		err := tx.Model(&models_main.StatUser{}).Where("source_key=? and date=?", user.RegSourceId, todayString).Count(&exist).Update("active_user", gorm.Expr("active_user+1")).Error
		if err != nil {
			tx.Rollback()
			s.logger.Debug("StatUserRegisterChannelHandler Error: ", err)
			return false
		}

		if exist == 0 {
			r := &models_main.StatUser{
				SourceKey:  user.RegSourceId,
				SourceName: channel.SourceName,
				ActiveUser: 1,
				Date:       today,
			}

			err = tx.Model(&models_main.StatUser{}).Create(&r).Error
			if err != nil {
				tx.Rollback()
				return false
			}
		}

		tx.Commit()

		global.GetRds().Set(key, 1, time.Second*time.Duration(timehelper.DateSkewing(1)-time.Now().Unix()))
	}
	return true
}

// StatUserRoomWatchHandler 直播间观看时长
func (s *ChannelStatistic) StatUserLiveWatchHandler(msg []byte) bool {
	var user struct {
		RoomId int
		models_main.User
	}
	err := json.Unmarshal(msg, &user)
	if err != nil {
		s.logger.Errorf("StatUserLiveWatchHandler err:%v", err)
		return false
	}

	//判断用户是否处于直播间页面
	structData, err := cache.GetLiveRoomUserInOutStatistics(user.RoomId, user.Id)
	if err != nil {
		return true
	}
	if structData.OutDeal {
		//用户已经退出直播间页面, 不计算时间
		return true
	}

	nowSec := time.Now().Unix()
	key := _const.LiveWatchDevice
	wd := cache.LiveWatchDevice{
		DeviceId: user.DeviceInfo,
		Duration: 0,
	}

	wds := global.GetRds().HGet(key, user.DeviceInfo).Val()
	if wds == "" {
		//记录最初观看时间到观众列表
		wd.StTime = nowSec
		wd.LastInstant = nowSec
		wd.Instant = nowSec
	} else {
		wd = *wd.Unmarshal(wds)
		if wd.IsOffline(nowSec) {
			//离线超过1分钟, 重新计算
			wd.StTime = nowSec
			wd.LastInstant = nowSec
			wd.Instant = nowSec
		} else {
			//记录最新的观看时间到观看时刻表
			wd.LastInstant = wd.Instant
			wd.Instant = nowSec
		}
	}
	wd.Duration += int(wd.Instant - wd.LastInstant)
	err = global.GetRds().HSet(key, user.DeviceInfo, wd.Marshal()).Err()
	if err != nil {
		s.logger.Debug("StatUserLiveWatchHandler Error: ", err)
		return false
	}

	if nowSec-wd.StTime > 0 {
		//需要更新的数据记录起来
		global.GetRds().HSet(_const.LiveWatchUpdateList, user.DeviceInfo, wd.Marshal())
	}
	return true
}

// StatUserRoomWatchHandler 直播间观看时长
func (s *ChannelStatistic) DyTransformRegisterHandler(msg []byte) bool {
	global.Log.Debugf("DyTransformRegisterHandler req: %+v", string(msg))
	trsReq := api_models.AscribeTransformReq{}
	err := json.Unmarshal(msg, &trsReq)
	if err != nil {
		global.Log.Errorf("DyTransformRegisterHandler err:%v", err)
		return false
	}
	record, err := report.NewAscribeReport().GetReportRecord(trsReq.AndroidId)
	if err != nil {
		global.Log.Errorf("GetReportRecord 查询失败: %v", err)
		return false
	}
	if record == nil {
		global.Log.Errorf("ReportRecord 不存在")
		return false
	}
	trsReq.Callback = record.Callback
	trsReq.CustomerActiveTime = time.Now().UnixMilli()
	ss := services.Service{}
	err = ss.SendAscribeTransform(&trsReq)
	if err != nil {
		global.Log.Debugf("巨量转化失败:%v", err)
		return false
	}
	return true
}
